package com.pccc.batch.config.remotepartition;

import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
import org.springframework.batch.integration.partition.BeanFactoryStepLocator;
import org.springframework.batch.integration.partition.StepExecutionRequestHandler;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpMethod;
import org.springframework.http.client.OkHttp3ClientHttpRequestFactory;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.http.converter.SerializingHttpMessageConverter;
import org.springframework.integration.http.dsl.Http;
import org.springframework.integration.http.support.DefaultHttpHeaderMapper;

import com.pccc.batch.config.remotepartition.model.Person;

@Configuration
@EnableBatchIntegration
@EnableBatchProcessing
public class WorkerConfiguration {

	@Autowired
    private ApplicationContext applicationContext;
	
	private final JobExplorer jobExplorer;
	
	private final StepBuilderFactory stepBuilderFactory;
	
	public WorkerConfiguration(StepBuilderFactory stepBuilderFactory,JobExplorer jobExplorer) {
		this.stepBuilderFactory = stepBuilderFactory;
		this.jobExplorer = jobExplorer;
	}
	
	
	@Bean(name = "slaveStep")
    public Step slaveStep(ItemReader<Person> reader, ItemWriter<Person> writer, ItemProcessor<Person, Person> processor) {
        return stepBuilderFactory.get("slaveStep")
                .<Person, Person>chunk(5)
                .reader(reader)
                .processor(processor)
                .writer(writer)
                .build();
    }
	
	
	@Bean
	public DirectChannel inputRequest() {
		return new DirectChannel();
	}
	
	
	@Bean
	public IntegrationFlow inboundFlow(DefaultHttpHeaderMapper httpMapper) {
		return IntegrationFlows
				.from(Http.inboundGateway("/store")
						  .requestChannel(inputRequest())
						  .requestMapping(m -> {
								m.consumes("application/x-java-serialized-object");
								m.methods(HttpMethod.POST);
							})
						  .headerMapper(httpMapper)
						  .messageConverters(new SerializingHttpMessageConverter()))
				.get();
	}
	
	
	
//	@Bean
//	public DirectChannel replyMaster() {
//		return new DirectChannel();
//	}
//    
//    @Bean
//    public IntegrationFlow httpReply(DefaultHttpHeaderMapper httpMapper) {
//        return IntegrationFlows.from(replyMaster())
//                .handle(Http.outboundGateway("http://localhost:8088/reply")
//                        .charset("UTF-8")
//                        .httpMethod(HttpMethod.POST)
//                        .messageConverters(new SerializingHttpMessageConverter())
//                        .headerMapper(httpMapper)
//                        .requestFactory(new OkHttp3ClientHttpRequestFactory()))
//                .log()
//                .get();
//    }
	
    
    @Bean
    public StepExecutionRequestHandler stepExecutionRequestHandler() {
        BeanFactoryStepLocator stepLocator = new BeanFactoryStepLocator();
        stepLocator.setBeanFactory(this.applicationContext);

        StepExecutionRequestHandler handler = new StepExecutionRequestHandler();
        handler.setStepLocator(stepLocator);
        handler.setJobExplorer(jobExplorer);
        return handler;
    }
	
}
